package com.yh.csx.etl.worker;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.druid.pool.DruidDataSource;
import com.google.common.base.Joiner;
import com.yh.csx.etl.config.EtlDataMapping;
import com.yh.csx.etl.config.EtlDataMapping.DbMapping;
import com.yh.csx.etl.config.EtlDataSourceManager;
import com.yh.csx.etl.support.EtlResult;
import com.yh.csx.etl.support.SplitTypeEnum;
import com.yh.csx.etl.util.DateUtils;
import com.yh.csx.etl.util.SqlUtils;
import com.yh.csx.etl.util.SyncUtils;

/***
 * etl处理程序
 * 
 * @author Robin.Wang
 * @date 2020-03-18
 */
public class EtlWorker implements Runnable {
    protected Logger log = LoggerFactory.getLogger(EtlWorker.class);

    private final long COUNT_PER_TASK = 10000L;
    /**
     * 同步配置
     */
    private EtlDataMapping dataMappingConfig;

    public EtlWorker(EtlDataMapping dataMappingConfig) {
        super();
        this.dataMappingConfig = dataMappingConfig;
    }

    @Override
    public void run() {
        this.importData(null);

    }

    /**
     * 导入数据入口，支持逻辑表多个物理表情况
     * 
     */
    public List<EtlResult> importData(List<String> params) {
        DbMapping dbMapping = dataMappingConfig.getDbMapping();
        String[] tables = dbMapping.getTable().split(",");
        List<EtlResult> results = new ArrayList<>();
        for (int i = 0; i < tables.length; i++) {
            EtlResult ret;
            try {
                ret = importData(tables[i], params);
            } catch (Exception e) {
                ret = new EtlResult();
                ret.setErrorMessage(e.getMessage());
                ret.setSucceeded(false);
            }
            results.add(ret);
        }
        return results;

    }

    /**
     * Etl 物理表数据
     * 
     * @throws Exception
     */
    public EtlResult importData(String sourceTable, List<String> params) throws Exception {
        DruidDataSource sourceDS = EtlDataSourceManager.DATA_SOURCES.get(dataMappingConfig.getDataSourceKey());
        DruidDataSource targetDS = EtlDataSourceManager.DATA_SOURCES.get(dataMappingConfig.getOuterSourceKey());
        String targetTable = SyncUtils.getTargetTableName(this.dataMappingConfig.getDbMapping(), sourceTable);
        syncTableDefination(sourceDS, targetDS, sourceTable, targetTable);
        String sql = "SELECT * FROM " + sourceTable;
        if (log.isDebugEnabled()) {
            log.debug("etl sql : {}", sql);
        }
        EtlResult ret = innerImportData(sourceDS, targetDS, sourceTable, sql, params);
        return ret;
    }

    /**
     * 
     * */
    private EtlResult innerImportData(DruidDataSource sourceDS, DruidDataSource targetDS, String sourceTable,
            String sql, List<String> params) {

        EtlResult etlResult = new EtlResult();
        AtomicLong impCount = new AtomicLong();
        List<String> errMsg = new ArrayList<>();
        List<String> delSqls = new ArrayList<String>();

        long start = System.currentTimeMillis();
        try {
            /**
             * keepDays: 30 splitType: monthly splitSuffixFormat: YYYY_MM
             */
            List<Object> paramValues = new ArrayList<>();
            // 拼接条件
            if (dataMappingConfig.getDbMapping().getEtlCondition() != null && params != null) {
                String etlCondition = dataMappingConfig.getDbMapping().getEtlCondition();
                for (String param : params) {
                    etlCondition = etlCondition.replace("{}", "?");
                    paramValues.add(param);
                }
                sql += " " + etlCondition;

            } else {
                if (SplitTypeEnum.SOLID.getValue()
                        .equals(dataMappingConfig.getDbMapping().getSplitType().toUpperCase())) {
                    String etlCondition = dataMappingConfig.getDbMapping().getEtlCondition();
                    etlCondition = etlCondition.replaceAll("\\{\\}", "?");
                    Date dateBeforePeriod = DateUtils
                            .getDateBeforePeriod(dataMappingConfig.getDbMapping().getKeepPeriod());
                    paramValues.add(dateBeforePeriod);
                    paramValues.add(DateUtils.getDateBeforePeriods(dateBeforePeriod,
                            dataMappingConfig.getDbMapping().getEtlPeriod()));
                    sql += " " + etlCondition;
                } else {
                    String etlCondition = dataMappingConfig.getDbMapping().getEtlCondition();
                    etlCondition = etlCondition.replaceAll("\\{\\}", "?");
                    Date dateBeforePeriod = DateUtils
                            .getDateBeforePeriod(dataMappingConfig.getDbMapping().getKeepPeriod());
                    paramValues.add(dateBeforePeriod);
                    paramValues.add(DateUtils.getDateBeforePeriods(dateBeforePeriod,
                            dataMappingConfig.getDbMapping().getEtlPeriod()));
                    sql += " " + etlCondition;
                }
            }
            // 获取总数
            String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
            long sourceCount = (Long) SqlUtils.sqlRS(sourceDS, countSql, paramValues, rs -> {
                Long count = null;
                try {
                    if (rs.next()) {
                        count = ((Number) rs.getObject(1)).longValue();
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
                return count == null ? 0L : count;
            });
            // 当大于1万条记录时多线程分页处理
            if (sourceCount >= COUNT_PER_TASK) {
                long offset;
                long size = COUNT_PER_TASK;
                long workerCnt = sourceCount / size + (sourceCount % size == 0 ? 0 : 1);
                for (long i = 0; i < workerCnt; i++) {
                    offset = size * i;
                    String sqlFinal = sql + " LIMIT " + offset + "," + size;
                    executeSqlImport(sourceDS, targetDS, sourceTable, sqlFinal, paramValues, impCount, delSqls, errMsg);
                }
            } else {
                executeSqlImport(sourceDS, targetDS, sourceTable, sql, paramValues, impCount, delSqls, errMsg);
            }
            log.info(sourceTable + "数据导入完成,总共{}, 一共导入 {} 条数据, 耗时: {}", sourceCount, impCount.get(),
                    System.currentTimeMillis() - start);
            etlResult.setResultMessage("导入 数据：" + impCount.get() + " 条");
        } catch (Exception e) {
            log.error("{} 数据导入异常{}", sourceTable, e);
            errMsg.add(sourceTable + " 数据导入异常 =>" + e.getMessage());
        }
        if (errMsg.isEmpty()) {
            etlResult.setSucceeded(true);
        } else {
            etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
        }
        return etlResult;
    }

    /**
     * 同步表结构
     */
    protected void syncTableDefination(DataSource sourceDS, DataSource targetDS, String sourceTable,
            String targetTable) {
        boolean isExistTable = (boolean) SqlUtils.sqlRS(targetDS, "show tables like '" + targetTable + "'", rs -> {
            try {// 如果表不存在则创建表
                return rs.next();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                return false;
            }
        });
        if (!isExistTable) {
            SqlUtils.sqlRS(sourceDS, "show create table " + sourceTable, rs -> {
                try {
                    if (rs.next()) {
                        String dml = rs.getString(2);
                        dml = dml.replace(sourceTable, targetTable);
                        SqlUtils.sqlExcute(targetDS, dml);// 创建表
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    return false;
                }
                return true;
            });
        } else {// 对比表字段是否佑修改
            DbMapping dbMapping = dataMappingConfig.getDbMapping();
            Map<String, String> targetColumnsMap = new LinkedHashMap<>();
            Map<String, Integer> targetColumnType = new LinkedHashMap<>();
            List<Map<String, Object>> pkAllIds = new LinkedList<Map<String, Object>>();

            try {
                SqlUtils.sqlRS(targetDS, "SELECT * FROM " + targetTable + " LIMIT 1 ", rs -> {
                    try {// 查询表定义及字段类型
                        ResultSetMetaData rsd = rs.getMetaData();
                        int columnCount = rsd.getColumnCount();
                        List<String> columns = new ArrayList<>();
                        for (int i = 1; i <= columnCount; i++) {
                            targetColumnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i));
                            columns.add(rsd.getColumnName(i));
                        }

                        targetColumnsMap.putAll(SyncUtils.getColumnsMap(dbMapping, columns));
                        return true;
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                        return false;
                    }
                });
                Map<String, String> sourceColumnsMap = new LinkedHashMap<>();
                Map<String, Integer> sourceColumnType = new LinkedHashMap<>();
                SqlUtils.sqlRS(sourceDS, "SELECT * FROM " + sourceTable + " LIMIT 1 ", rs -> {
                    try {// 查询表定义及字段类型
                        ResultSetMetaData rsd = rs.getMetaData();
                        int columnCount = rsd.getColumnCount();
                        List<String> columns = new ArrayList<>();
                        for (int i = 1; i <= columnCount; i++) {
                            sourceColumnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i));
                            columns.add(rsd.getColumnName(i));
                        }

                        sourceColumnsMap.putAll(SyncUtils.getColumnsMap(dbMapping, columns));
                        return true;
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                        return false;
                    }
                });
            } catch (Exception e) {
                log.error("同步表结构失败:{}", e);
            }
        }
    }

    /**
     * 执行导入
     */
    protected boolean executeSqlImport(DataSource sourceDS, DataSource targetDS, String sourceTable, String finalSql,
            List<Object> paramValues, AtomicLong impCount, List<String> delSqls, List<String> errMsg) {
        try {
            DbMapping dbMapping = dataMappingConfig.getDbMapping();
            Map<String, String> columnsMap = new LinkedHashMap<>();
            Map<String, Integer> columnType = new LinkedHashMap<>();
            List<Map<String, Object>> pkAllIds = new LinkedList<Map<String, Object>>();

            SqlUtils.sqlRS(targetDS,
                    "SELECT * FROM " + SyncUtils.getTargetTableName(dbMapping, sourceTable) + " LIMIT 1 ", rs -> {
                        try {// 查询表定义及字段类型
                            ResultSetMetaData rsd = rs.getMetaData();
                            int columnCount = rsd.getColumnCount();
                            List<String> columns = new ArrayList<>();
                            for (int i = 1; i <= columnCount; i++) {
                                columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i));
                                columns.add(rsd.getColumnName(i));
                            }

                            columnsMap.putAll(SyncUtils.getColumnsMap(dbMapping, columns));
                            return true;
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                            return false;
                        }
                    });

            SqlUtils.sqlRS(sourceDS, finalSql, paramValues, rs -> {
                int idx = 1;
                try {
                    boolean completed = false;

                    StringBuilder insertSql = new StringBuilder();
                    insertSql.append("REPLACE INTO ").append(SyncUtils.getTargetDbTableName(dbMapping, sourceTable))
                            .append(" (");
                    columnsMap.forEach(
                            (targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));// 拼接字段
                    int len = insertSql.length();
                    insertSql.delete(len - 1, len).append(") VALUES (");
                    int mapLen = columnsMap.size();
                    for (int i = 0; i < mapLen; i++) {// 拼接参数
                        insertSql.append("?,");
                    }
                    len = insertSql.length();
                    insertSql.delete(len - 1, len).append(")");
                    // sql执行插入
                    try (Connection connTarget = targetDS.getConnection();
                            PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
                        connTarget.setAutoCommit(false);
                        List<Object> pkBatchIds = new ArrayList<Object>();
                        while (rs.next()) {
                            completed = false;
                            pstmt.clearParameters();

                            int i = 1;
                            for (Map.Entry<String, String> entry : columnsMap.entrySet()) {// 拼接值
                                String targetClolumnName = entry.getKey();
                                String srcColumnName = entry.getValue();
                                if (srcColumnName == null) {
                                    srcColumnName = targetClolumnName;
                                }

                                Integer type = columnType.get(targetClolumnName.toLowerCase());
                                Object value = rs.getObject(srcColumnName);
                                if (value != null) {
                                    SyncUtils.setPStmt(type, pstmt, value, i);
                                } else {
                                    pstmt.setNull(i, type);
                                }
                                i++;
                            }

                            pstmt.execute();
                            if (log.isTraceEnabled()) {
                                log.trace("replace into target table, sql: {}", insertSql);
                            }
                            if (idx % dbMapping.getCommitBatch() == 0) {// 批量提交事务
                                connTarget.commit();
                                pkBatchIds.clear();
                                completed = true;
                            }
                            Map<String, Object> pkVal = buildKeyVal(dbMapping, rs);
                            // String key=buildKey(pkVal);
                            pkAllIds.add(pkVal);
                            idx++;
                            impCount.incrementAndGet();
                            if (log.isDebugEnabled()) {
                                log.debug("successful import count:" + impCount.get());
                            }
                        }
                        if (!completed) {// 事务提交
                            connTarget.commit();
                            completed = true;
                        }
                    }
                } catch (Exception e) {
                    log.error(dbMapping.getTable() + ":" + sourceTable + " etl failed! ==>" + e.getMessage(), e);
                    errMsg.add(dbMapping.getTable() + " etl failed! ==>" + e.getMessage());
                }
                return idx;
            });

            // 删除数据
            int delCount = 0;
            boolean delCompleted = false;
            try (Connection srcConn = sourceDS.getConnection()) {
                srcConn.setAutoCommit(false);
                PreparedStatement psmt = srcConn.prepareStatement("SET sql_log_bin=0;");// 不记录Binlog日志
                psmt.execute();
                for (Map<String, Object> entry : pkAllIds) {
                    StringBuilder deleteSql = new StringBuilder("DELETE FROM " + sourceTable + " WHERE ");
                    appendCondition(dbMapping, deleteSql, entry);
                    try (PreparedStatement pstmt2 = srcConn.prepareStatement(deleteSql.toString())) {
                        int k = 1;
                        for (Object val : entry.values()) {

                            pstmt2.setObject(k++, val);
                        }
                        pstmt2.execute();
                        if (delCount % dbMapping.getCommitBatch() == 0) {
                            srcConn.commit();
                            delCompleted = true;
                        } else {
                            delCompleted = false;
                        }
                    }
                    delCount++;
                }
                if (!delCompleted) {
                    srcConn.commit();
                }
            }
            return true;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return false;
        }
    }

    /**
     * 拼接目标表主键where条件
     */
    private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map<String, Object> values)
            throws SQLException {
        // 拼接主键
        for (Map.Entry<String, String> entry : dbMapping.getTargetPk().entrySet()) {
            sql.append(entry.getKey()).append("=? AND ");
        }
        sql.delete(sql.length() - 5, sql.length() - 1);
    }

    public static Map<String, Map<String, String>> pkCondations(DbMapping dbMapping, StringBuilder sql,
            Map<String, Object> values, ResultSet rs) throws SQLException {
        // 拼接主键
        for (Map.Entry<String, String> entry : dbMapping.getTargetPk().entrySet()) {
            String targetColumnName = entry.getKey();
            String srcColumnName = entry.getValue();
            if (srcColumnName == null) {
                srcColumnName = targetColumnName;
            }
            sql.append(targetColumnName).append("=? AND ");
            values.put(targetColumnName, rs.getObject(srcColumnName));
        }
        int len = sql.length();
        sql.delete(len - 4, len);

        return null;

    }

    public Map<String, Object> buildKeyVal(DbMapping dbMapping, ResultSet rs) throws SQLException {
        Map<String, Object> values = new LinkedHashMap<>();
        for (Map.Entry<String, String> entry : dbMapping.getTargetPk().entrySet()) {
            String targetColumnName = entry.getKey();
            String srcColumnName = entry.getValue();
            if (srcColumnName == null) {
                srcColumnName = targetColumnName;
            }
            values.put(targetColumnName, rs.getObject(srcColumnName));
        }
        return values;

    }

    public static String buildKey(Map<String, Object> pkVals) {
        StringBuilder sb = new StringBuilder();
        for (Map.Entry<String, Object> entry : pkVals.entrySet()) {
            sb.append(entry.getKey()).append(":").append(entry.getValue().toString()).append(",");
        }
        return sb.toString();
    }

}
